package com.software.process.now

import com.software.util.DBTools
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{SaveMode}
import org.apache.spark.sql.functions.mean

object AQIMeanCalculator {
  Logger.getLogger("org").setLevel(Level.ERROR)
  def main(args: Array[String]): Unit = {

    val spark = DBTools.getSession("AirQuailty","local")
    val df = DBTools.redFile("csv","D:\\csv\\date.csv",spark)

    val filterByAQI = df.filter(df("type") === "AQI")

    // 选择从第三列开始的所有列
    val colsToAvg = filterByAQI.columns.slice(3, df.columns.length)

    // 计算每列的平均值
    val result = colsToAvg.map(col => {
      val avg = filterByAQI.select(mean(col)).first.getDouble(0)
      (col, avg)
    })


    // 将结果转换为DataFrame，并保存到数据库中
    val resultDF = spark.createDataFrame(result).toDF("city", "value")
    resultDF.write.mode(SaveMode.Overwrite)
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/AirDB?serverTimezone=GMT%2B8")
      .option("driver", "com.mysql.cj.jdbc.Driver")
      .option("user", "root")
      .option("password", "lyf20020511")
      .option("dbtable", "aqi_mean_value")
      .mode(SaveMode.Append)
      .save()

    // 停止SparkSession对象
    spark.stop()
  }
}